package com.ichoice.rabbitmq.rpc.helper;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.ClassUtil;
import cn.hutool.core.util.ReflectUtil;
import com.ichoice.rabbitmq.rpc.helper.client.ClientBlinding;
import com.ichoice.rabbitmq.rpc.helper.consts.CommonContant;
import com.ichoice.rabbitmq.rpc.helper.model.CallBackModel;
import com.ichoice.rabbitmq.rpc.helper.model.DelayMessagePoster;
import com.ichoice.rabbitmq.rpc.helper.model.ProducerModel;
import com.ichoice.rabbitmq.rpc.helper.model.SubScribeModel;
import com.ichoice.rabbitmq.rpc.helper.sub.SubScribeStoreHelper;
import com.ichoice.rabbitmq.rpc.helper.util.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.logging.log4j.util.Strings;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.lang.reflect.Method;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.Collection;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;

/**
 * @author xianrensawang
 * @since 2021/1/28
 */
@Slf4j
public class RpcRabbitContext extends RpcRabbitAdapter implements RpcRabbitContextAware {

    private final RabbitTemplate rabbitTemplate;

    private final ClientBlinding blinding;

    private String exchange;
    private String backQueue;
    private String backKey;

    private String applicationNameCache;
    private final ConcurrentHashMap<Long, CallBackModel> returnResultCache;
    private final Long timeOut;
    private final ThreadPoolTaskExecutor taskExecutor;

    public RpcRabbitContext(RabbitTemplate rabbitTemplate, ClientBlinding blinding, long time) {
        this.rabbitTemplate = rabbitTemplate;
        this.blinding = blinding;
        timeOut = time;
        returnResultCache = new ConcurrentHashMap<>();
        // 发送端接收消费端回参线程池
        taskExecutor = ExecutorUtils.createPool("client-receive");
    }

    /**
     * 初始化绑定返回队列
     */
    public void init(String applicationName, String serverPort) throws UnknownHostException {
        applicationNameCache = applicationName;
        String host = InetAddress.getLocalHost().getHostAddress();
        // 绑定公共参数
        CommonContant.applicationName = applicationName;
        CommonContant.ipAddress = host;
        CommonContant.port = serverPort;
        backQueue = MqMetadataUtils.selfBackQueue(applicationName, host, serverPort);
        CommonContant.backQueue = backQueue;
        backKey = MqMetadataUtils.selfBackKey(applicationName, host, serverPort);
        exchange = CommonContant.EXCHANGE;
        // 绑定队列监听方法
        blinding.blind(exchange, backKey, backQueue);
        // 绑定广播配置
        blinding.blindFanout(exchange, backQueue, MqMetadataUtils.fanoutKey(applicationName));
    }


    @Override
    public void sentImpl(ProducerModel producerModel) {
        completeModel(producerModel, null, "", Boolean.FALSE);
        sentMessage(producerModel);
    }

    @Override
    public void sentAndBackImpl(ProducerModel producerModel, ClientListenServerCallback callback) {
        completeModel(producerModel, callback, "", Boolean.TRUE);
        // 发送消息
        sentMessage(producerModel);
    }

    @Override
    public void sentBackByTargetMethodImpl(ProducerModel producerModel, ClientListenServerCallback callback, String method) {
        completeModel(producerModel, callback, method, Boolean.TRUE);
        // 发送消息
        sentMessage(producerModel);
    }

    @Override
    public void consumer(CallBackModel callBackModel) throws Exception {
        // 错误返回结果,抛出异常
        if (Objects.nonNull(callBackModel) && Objects.equals(callBackModel.getCode(), 500)) {
            // 抛出异常信息
            throw new Exception(callBackModel.getMessage());
        }
        try {
            if (returnResultCache.containsKey(callBackModel.getAck())) {
                returnResultCache.put(callBackModel.getAck(), callBackModel);
            } else {
                // 无回调bean
                if (Strings.isBlank(callBackModel.getCallBackBean())) {
                    return;
                }
                Class<?> aClass = Class.forName(callBackModel.getCallBackBean());
                Object bean = SpringUtil.getBean(aClass);
                String method = "listenServerCallback";
                if (Strings.isNotBlank(callBackModel.getCallBackMethod())) {
                    method = callBackModel.getCallBackMethod();
                }
                Object backArg = null;
                Method[] methods = ReflectUtil.getMethods(aClass);
                for (Method m : methods) {
                    if (Objects.equals(m.getName(), method)) {
                        Class<?>[] parameterTypes = m.getParameterTypes();
                        if (parameterTypes.length != 1) {
                            throw new IllegalArgumentException("回调函数有且只有一个传参");
                        }
                        if (parameterTypes[0].equals(CallBackModel.class)) {
                            backArg = callBackModel;
                        } else {
                            boolean primitiveWrapper = ClassUtil.isPrimitiveWrapper(parameterTypes[0]);
                            boolean basicType = ClassUtil.isBasicType(parameterTypes[0]);
                            boolean isString = String.class.isAssignableFrom(parameterTypes[0]);
                            boolean isCollection = Collection.class.isAssignableFrom(parameterTypes[0]);
                            if (!primitiveWrapper && !basicType && !isCollection && !isString) {
                                Object instance = parameterTypes[0].newInstance();
                                BeanUtil.copyProperties(callBackModel.getRes(), instance);
                                backArg = instance;
                            } else {
                                backArg = callBackModel.getRes();
                            }
                        }
                        break;
                    }
                }
                ReflectUtil.invoke(bean, method, backArg);
            }
        } catch (Exception e) {
            log.error("客户端执行消息队列rpc回调方法出现异常,ackId:{},客户端:{},服务端:{},回调类:{},回调函数名:{},异常信息:{}",
                    callBackModel.getAck(), callBackModel.getCurrentApplication(), callBackModel.getTargetApplication(),
                    callBackModel.getCallBackBean(), callBackModel.getCallBackMethod(), e.getMessage());
            throw e;
        }
    }

    private void noRes(ProducerModel producerModel) {
        long ack = AckIdUtil.nextId();
        producerModel.setAck(ack);
        producerModel.setNeedRes(Boolean.FALSE);
    }


    private void completeModel(ProducerModel producerModel, ClientListenServerCallback callback, String method, Boolean needRes) {
        long ack = AckIdUtil.nextId();
        if (Objects.nonNull(callback)) {
            String name = callback.getClass().getName();
            producerModel.setCallBackBean(name);
        }
        producerModel.setAck(ack);
        if (needRes) {
            producerModel.setCallBackQueue(backQueue);
            producerModel.setCallBackRoutKey(backKey);
            producerModel.setCallBackExchange(exchange);
        }
        producerModel.setNeedRes(needRes);
        if (Strings.isNotBlank(method)) {
            producerModel.setCallBackMethod(method);
        }
    }


    @Override
    public void fanout(Object... args) {
        StackTraceElement[] stack = Thread.currentThread().getStackTrace();
        StackTraceElement element = stack[2];
        // 获取调用方法名
        String methodName = element.getMethodName();
        // 获取调用类名
        String className = StringUit.getClassName(element.getClassName());
        // 拼接 发布订阅关系缓存 cacheKey
        String cacheKey = StringUit.firstCharLower(className) + "-" + methodName;
        // 获取发布订阅关系列表
        Map<String, SubScribeModel> scribeModelMap = SubScribeStoreHelper.get(cacheKey);
        if (CollectionUtil.isNotEmpty(scribeModelMap)) {
            scribeModelMap.forEach((key, value) -> {
                // 生成请求客户端参数
                ProducerModel producerModel = ProducerModel.builder()
                        .targetApplication(key)
                        .bean(value.getClientBean())
                        .method(value.getClientMethod())
                        .args(args)
                        .build();
                // 发送消息到订阅的客户端
                sentImpl(producerModel);
                log.info("广播消息到{}客户端成功，客户端bean:{},客户端方法:{},消息:{}", key, value.getClientBean(),
                        value.getClientMethod(), Arrays.toString(args));
            });
        } else {
            log.info("通过key获取发布订阅列表无数据,key:{}", cacheKey);
        }
    }

    @Override
    public CallBackModel sentAndBack(ProducerModel producerModel) {
        CallBackModel finalResult = null;
        Future<CallBackModel> future = taskExecutor.submit(() -> {
            completeModel(producerModel, null, "", Boolean.TRUE);
            Long ack = producerModel.getAck();
            CallBackModel res = CallBackModel.fail(-1L, "未获取到回参");
            returnResultCache.put(ack, res);
            sentMessage(producerModel);
            boolean isBreak = true;
            long start = System.currentTimeMillis();
            while (isBreak && System.currentTimeMillis() - start < timeOut) {
                res = returnResultCache.get(ack);
                isBreak = Objects.equals(res.getAck(), -1L);
            }
            returnResultCache.remove(ack);
            return res;
        });
        try {
            finalResult = future.get();
        } catch (Exception e) {
            log.error("获取future结果失败:{}", e.getMessage());
        }
        return finalResult;
    }


    /**
     * 实时发送
     *
     * @param producerModel 入参
     */
    private void sentMessage(ProducerModel producerModel) {
        // 无延迟
        if (Objects.isNull(producerModel.getTtl())) {
            noDelaySentMessage(producerModel);
        } else {
            // 指定延迟发送
            delaySentMessage(producerModel);
        }
    }

    /**
     * 实时发送
     *
     * @param producerModel 入参
     */
    private void noDelaySentMessage(ProducerModel producerModel) {
        String target = producerModel.getTargetApplication();
        String exchange = CommonContant.EXCHANGE;
        // 生成目标服务路由key
        String key = MqMetadataUtils.selfKey(target);
        producerModel.setCurrentApplication(applicationNameCache);
        rabbitTemplate.convertAndSend(exchange, key, producerModel);
    }

    /**
     * 消息延迟发送
     *
     * @param producerModel 入参
     */
    private void delaySentMessage(ProducerModel producerModel) {
        String target = producerModel.getTargetApplication();
        String exchange = CommonContant.EXCHANGE_DELAY;
        // 生成目标服务延迟队列路由key
        String key = MqMetadataUtils.selfDelayKey(target);
        producerModel.setCurrentApplication(applicationNameCache);
        rabbitTemplate.convertAndSend(exchange, key, producerModel, new DelayMessagePoster(producerModel.getTtl()));
    }

}
